package com.lintf.distributed.transaction.order.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.lintf.distributed.transaction.DeductInventoryEvent;
import com.lintf.distributed.transaction.MessageProcessedEvent;
import com.lintf.distributed.transaction.order.entity.Message;
import com.lintf.distributed.transaction.order.entity.Order;
import com.lintf.distributed.transaction.order.service.MessageService;
import com.lintf.distributed.transaction.order.service.OrderService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Component
@RequiredArgsConstructor
public class MessageResultListener {

    private final MessageService messageService;
    private final OrderService orderService;

    @KafkaListener(topics = "message-processed",
            groupId = "order-service-group")
    @Transactional(rollbackFor = Exception.class)
    public void handleMessageResult(ConsumerRecord<String, String> record) {
        String messageId = record.key();
        String json = record.value();

        log.info("收到消息处理结果: key={}", messageId);

        try {
            MessageProcessedEvent event = JSON.parseObject(json, MessageProcessedEvent.class);

            LambdaUpdateWrapper<Message> messageUpdateWrapper = new LambdaUpdateWrapper<>();
            messageUpdateWrapper.eq(Message::getMessageId, messageId)
                    .set(Message::getStatus, event.isSuccess() ? 2 : 3)
                    .set(Message::getRemark, event.getReason())
                    .set(Message::getUpdatedAt, event.getProcessedAt());
            messageService.update(messageUpdateWrapper);

            DeductInventoryEvent deductInventoryEvent = JSON.parseObject(event.getContent(), DeductInventoryEvent.class);
            LambdaUpdateWrapper<Order> orderUpdateWrapper = new LambdaUpdateWrapper<>();
            if (event.isSuccess()) {
                log.info("消息处理成功: messageId={}", messageId);
                orderUpdateWrapper.eq(Order::getId, deductInventoryEvent.getOrderId())
                        .set(Order::getStatus, "INVENTORY_DEDUCTED")
                        .set(Order::getUpdatedAt, event.getProcessedAt());
            } else {
                log.warn("消息处理失败: messageId={}, reason={}", messageId, event.getReason());
                orderUpdateWrapper.eq(Order::getId, deductInventoryEvent.getOrderId())
                        .set(Order::getStatus, "INVENTORY_FAILED")
                        .set(Order::getRemark, event.getReason())
                        .set(Order::getUpdatedAt, event.getProcessedAt());
            }
            orderService.update(orderUpdateWrapper);
        } catch (Exception e) {
            log.error("处理消息结果失败: messageId={}", messageId, e);
        }
    }
}